/**
 * Copyright 2016 vip.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 * </p>
 **/

package spring.cloud.tasks.api.service.impl;

import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import spring.cloud.tasks.api.database.entity.TaskConfigurationDto;
import spring.cloud.tasks.api.database.service.NameSpaceZooKeeperClusterMappingDtoService;
import spring.cloud.tasks.api.database.service.TaskConfigurationDtoService;
import spring.cloud.tasks.api.domain.*;
import spring.cloud.tasks.api.service.TaskService;
import spring.cloud.tasks.api.service.ZooKeeperDatabaseTaskDifferentInfoService;
import spring.cloud.tasks.api.service.ZooKeeperService;
import spring.cloud.tasks.api.utils.ConsoleThreadFactory;
import spring.cloud.tasks.api.utils.TaskNodePath;
import spring.cloud.tasks.api.zookeeper.ZooKeeper;
import spring.cloud.tasks.api.zookeeper.ZooKeeperRepository;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class ZooKeeperDatabaseDiffServiceImpl implements ZooKeeperDatabaseTaskDifferentInfoService {

    private static final Logger log = LoggerFactory.getLogger(ZooKeeperDatabaseDiffServiceImpl.class);

    private static final String NAMESPACE_NOT_EXIST_TEMPLATE = "The namespace {%s} does not exists.";

    private static final String ERR_MSG_SKIP_DIFF = "skip diff by namespace:{} for reason:{}";

    private static final int DIFF_THREAD_NUM = 10;

    @Resource
    private NameSpaceZooKeeperClusterMappingDtoService nameSpaceZooKeeperClusterMappingDtoService;

    @Resource
    private TaskConfigurationDtoService taskConfigurationDtoService;

    @Resource
    private TaskService taskService;

    @Resource
    private ZooKeeperService zooKeeperService;

    @Resource
    private ZooKeeperRepository zooKeeperRepository;

    private ExecutorService diffExecutorService;

    @PostConstruct
    public void init() {
        if (diffExecutorService != null) {
            diffExecutorService.shutdownNow();
        }
        diffExecutorService = Executors.newFixedThreadPool(DIFF_THREAD_NUM, new ConsoleThreadFactory("diff-zk-db-thread", false));
    }

    @PreDestroy
    public void destroy() {
        if (diffExecutorService != null) {
            diffExecutorService.shutdownNow();
        }
    }

    @Override
    public List<TaskDifferentInfo> differentByZooKeeperClusterKey(String zooKeeperClusterKey) throws RuntimeException {

        long startTime = System.currentTimeMillis();
        List<String> nameSpaceList = nameSpaceZooKeeperClusterMappingDtoService.getNameSpaceList(zooKeeperClusterKey);

        List<Callable<List<TaskDifferentInfo>>> callableList = Lists.newArrayList();
        for (final String nameSpace : nameSpaceList) {
            Callable<List<TaskDifferentInfo>> callable = new Callable<List<TaskDifferentInfo>>() {
                @Override
                public List<TaskDifferentInfo> call() throws Exception {
                    return differentByNameSpace(nameSpace);
                }
            };
            callableList.add(callable);
        }

        List<TaskDifferentInfo> diffInfoList = Lists.newArrayList();
        try {
            List<Future<List<TaskDifferentInfo>>> futureList = diffExecutorService.invokeAll(callableList);

            for (Future<List<TaskDifferentInfo>> future : futureList) {
                List<TaskDifferentInfo> taskDiffInfoList = future.get();
                if (taskDiffInfoList != null && !taskDiffInfoList.isEmpty()) {
                    diffInfoList.addAll(taskDiffInfoList);
                }
            }
        } catch (InterruptedException e) {// NOSONAR
            log.warn("the thread is interrupted", e);
            throw new RuntimeException("the diff thread is interrupted", e);
        } catch (Exception e) {
            log.error("exception happens during execute diff operation", e);
            throw new RuntimeException(e);
        }

        log.info("Finish diff zkcluster:{}, which cost {}ms", zooKeeperClusterKey, System.currentTimeMillis() - startTime);

        return diffInfoList;
    }

    @Override
    public List<TaskDifferentInfo> differentByNameSpace(String nameSpace) throws RuntimeException {
        long startTime = System.currentTimeMillis();

        List<TaskDifferentInfo> taskDiffInfoArrayList = Lists.newArrayList();

        try {
            List<TaskConfigurationDto> taskConfigurationDtoList = taskConfigurationDtoService.getTaskConfigurationDtoListByNamespace(nameSpace);
            if (taskConfigurationDtoList == null || taskConfigurationDtoList.isEmpty()) {
                return taskDiffInfoArrayList;
            }
            ZooKeeper zooKeeper = initNameSpaceClient(nameSpace);
            if (zooKeeper == null) {
                return taskDiffInfoArrayList;
            }
            for (TaskConfigurationDto taskConfigurationDto : taskConfigurationDtoList) {
                String taskId = taskConfigurationDto.getTaskId();
                log.info("start to diff job:{}@{}", taskId, nameSpace);
                if (!checkTaskIsExsitInZooKeeper(taskId, zooKeeper)) {
                    taskDiffInfoArrayList.add(new TaskDifferentInfo(nameSpace, taskId, DifferentType.DATABASE_ONLY, Lists.<TaskConfigurationDifferentInfo>newArrayList()));
                    continue;
                }
                TaskConfiguration taskConfiguration = taskService.getTaskConfigurationFromZooKeeper(nameSpace, taskId);
                TaskDifferentInfo taskDiffInfo = diff(nameSpace, taskConfigurationDto, taskConfiguration, false);
                if (taskDiffInfo != null) {
                    taskDiffInfoArrayList.add(taskDiffInfo);
                }
            }
            Set<String> taskIdSetInDb = getTaskIdSetInDb(taskConfigurationDtoList);
            List<TaskDifferentInfo> jobsInZkOnly = getTaskIdOnlyInZooKeeper(nameSpace, taskIdSetInDb);
            if (jobsInZkOnly != null && !jobsInZkOnly.isEmpty()) {
                taskDiffInfoArrayList.addAll(jobsInZkOnly);
            }

        } catch (RuntimeException e) {
            log.error(e.getMessage(), e);
            throw e;
        } catch (Exception e) {
            log.error("exception throws during diff by namespace [{}]", nameSpace, e);
            throw new RuntimeException(e);
        } finally {
            log.info("Finish diff namespace:{} which cost {}ms", nameSpace, System.currentTimeMillis() - startTime);
        }

        return taskDiffInfoArrayList;
    }

    @Override
    public TaskDifferentInfo differentByTaskId(String nameSpace, String jobName) throws RuntimeException {
        ZooKeeper zooKeeper;
        try {
            zooKeeper = initNameSpaceClient(nameSpace);
            if (zooKeeper == null) {
                return null;
            }
            log.info("start to diff job:{}", jobName);

            TaskConfigurationDto taskConfiguration4DB = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, jobName);
            TaskConfiguration taskConfiguration = taskService.getTaskConfigurationFromZooKeeper(nameSpace, jobName);

            if (taskConfiguration4DB == null) {
                if (taskConfiguration != null) {
                    return new TaskDifferentInfo(nameSpace, jobName, DifferentType.ZOOKEEPER_ONLY,
                            Lists.<TaskConfigurationDifferentInfo>newArrayList());
                } else {
                    return null;
                }
            }

            if (taskConfiguration == null) {
                return new TaskDifferentInfo(nameSpace, jobName, DifferentType.DATABASE_ONLY,
                        Lists.<TaskConfigurationDifferentInfo>newArrayList());
            }

            // diff only when dbJobConfig and zkJobConfig both not null
            return diff(nameSpace, taskConfiguration4DB, taskConfiguration, true);
        } catch (Exception e) {
            log.error("exception throws during diff by namespace [{}] and job [{}]", nameSpace, jobName, e);
            throw new RuntimeException(e);
        }
    }

    private boolean checkTaskIsExsitInZooKeeper(String taskId, ZooKeeper zooKeeper) {
        return zooKeeper.checkExists(TaskNodePath.getTaskNodePath(taskId));
    }

    /**
     * zk中的作业配置和db中的作业配置的对比。
     *
     * @param dbTaskConfiguration db里面的配置。
     * @param zkTaskConfiguration zk里面的配置
     * @param needDetail          是否需要细节；true，则需要，false，为不需要；
     */
    protected TaskDifferentInfo diff(String namespace, TaskConfiguration dbTaskConfiguration, TaskConfiguration zkTaskConfiguration, boolean needDetail) {
        String taskId = dbTaskConfiguration.getTaskId();

        List<TaskConfigurationDifferentInfo> taskConfigurationDiffInfoList = Lists.newArrayList();
        // jobType
        diff("jobType", dbTaskConfiguration.getTaskType(), zkTaskConfiguration.getTaskType(), taskConfigurationDiffInfoList);
        // jobClass
        diff("jobClass", dbTaskConfiguration.getTaskClass(), zkTaskConfiguration.getTaskClass(), taskConfigurationDiffInfoList);
        // shardingTotalCount
        diff("shardingTotalCount", dbTaskConfiguration.getTaskItemTotalCount(), zkTaskConfiguration.getTaskItemTotalCount(), taskConfigurationDiffInfoList);
        // cron
        diff("cron", dbTaskConfiguration.getCronExpression(), zkTaskConfiguration.getCronExpression(), taskConfigurationDiffInfoList);
        // pausePeriodDate
        diff("pausePeriodDate", dbTaskConfiguration.getPauseDatePeriod(), zkTaskConfiguration.getPauseDatePeriod(), taskConfigurationDiffInfoList);
        // pausePeriodTime
        diff("pausePeriodTime", dbTaskConfiguration.getPauseTimePeriod(), zkTaskConfiguration.getPauseTimePeriod(), taskConfigurationDiffInfoList);
        // shardingItemParameters
        diff("shardingItemParameters", dbTaskConfiguration.getTaskItemParameters(), zkTaskConfiguration.getTaskItemParameters(), taskConfigurationDiffInfoList);
        // jobParameter
        diff("jobParameter", dbTaskConfiguration.getTaskParameter(), zkTaskConfiguration.getTaskParameter(), taskConfigurationDiffInfoList);
        // processCountIntervalSeconds
        diff("processCountIntervalSeconds", dbTaskConfiguration.getProcessCountIntervalSeconds(), zkTaskConfiguration.getProcessCountIntervalSeconds(), taskConfigurationDiffInfoList);
        // timeout4AlarmSeconds
        diff("timeout4AlarmSeconds", dbTaskConfiguration.getTimeout4AlarmSeconds(), zkTaskConfiguration.getTimeout4AlarmSeconds(), taskConfigurationDiffInfoList);
        // timeoutSeconds
        diff("timeoutSeconds", dbTaskConfiguration.getTimeoutSeconds(), zkTaskConfiguration.getTimeoutSeconds(), taskConfigurationDiffInfoList);
        // loadLevel
        diff("loadLevel", dbTaskConfiguration.getLoadLevel(), zkTaskConfiguration.getLoadLevel(), taskConfigurationDiffInfoList);
        // enabled
        diff("enabled", dbTaskConfiguration.getEnabled(), zkTaskConfiguration.getEnabled(), taskConfigurationDiffInfoList);
        // preferList
        diff("preferList", dbTaskConfiguration.getPreferList(), zkTaskConfiguration.getPreferList(), taskConfigurationDiffInfoList);
        // useDispreferList
        diff("useDispreferList", dbTaskConfiguration.getUseDisPreferList(), zkTaskConfiguration.getUseDisPreferList(), taskConfigurationDiffInfoList);

        // description
        diff("description", dbTaskConfiguration.getDescription(), zkTaskConfiguration.getDescription(), taskConfigurationDiffInfoList);

        // showNormalLog
        diff("showNormalLog", dbTaskConfiguration.getShowNormalLog(), zkTaskConfiguration.getShowNormalLog(), taskConfigurationDiffInfoList);
        // showNormalLog
        diff("showNormalLog", dbTaskConfiguration.getShowNormalLog(), zkTaskConfiguration.getShowNormalLog(), taskConfigurationDiffInfoList);

        if (!taskConfigurationDiffInfoList.isEmpty()) {
            if (needDetail) {
                return new TaskDifferentInfo(namespace, taskId, DifferentType.HAS_DIFFERENCE, taskConfigurationDiffInfoList);
            }
            return new TaskDifferentInfo(namespace, taskId, DifferentType.HAS_DIFFERENCE, Lists.<TaskConfigurationDifferentInfo>newArrayList());
        }

        return null;
    }

    public void diff(String key, Object valueInDb, Object valueInZk, List<TaskConfigurationDifferentInfo> configDiffInfos) {
        // 这里处理所有valueInDB 为空的情况
        if (valueInDb == null) {
            if (valueInZk == null) {
                return;
            }

            // 空串与null视为相等
            if (valueInZk instanceof String && StringUtils.isEmpty((String) valueInZk)) {
                return;
            }

            log.debug("key:{} has difference between zk and db", key);
            configDiffInfos.add(new TaskConfigurationDifferentInfo(key, valueInDb, valueInZk));
            return;
        }

        // valueInDB != null && valueInZk == null
        if (valueInZk == null) {
            log.debug("key:{} has difference between zk and db", key);
            configDiffInfos.add(new TaskConfigurationDifferentInfo(key, valueInDb, valueInZk));
            return;
        }

        /** 下面情况 valueInDB and valueInZk 均非空 **/
        // 处理String类型
        if (valueInDb instanceof String) {
            String dbStr = (String) valueInDb;
            String zkStr = (String) valueInZk;

            if (StringUtils.isEmpty(dbStr) && StringUtils.isEmpty(zkStr)) {
                return;
            }

            if (!dbStr.trim().equals(zkStr.trim())) {
                log.debug("key:{} has difference between zk and db", key);
                configDiffInfos.add(new TaskConfigurationDifferentInfo(key, dbStr, zkStr));
                return;
            }

            return;
        }

        // 处理非String类型
        if (!valueInDb.equals(valueInZk)) {
            log.debug("key:{} has difference between zk and db", key);
            configDiffInfos.add(new TaskConfigurationDifferentInfo(key, valueInDb, valueInZk));
        }
    }

    private ZooKeeper initNameSpaceClient(String nameSpace) {
        NameSpaceConfiguration nameSpaceConfiguration = zooKeeperService.getNameSpaceConfigurationByNameSpace(nameSpace);
        if (nameSpaceConfiguration == null) {
            String errMsg = String.format(NAMESPACE_NOT_EXIST_TEMPLATE, nameSpace);
            log.warn(ERR_MSG_SKIP_DIFF, nameSpace, errMsg);
            return null;
        }
        NameSpaceClient nameSpaceClient = zooKeeperService.connectByNameSpace(nameSpace);
        if (nameSpaceClient != null && nameSpaceClient.isConnected()) {
            return zooKeeperRepository.getZooKeeper(nameSpaceClient.getCuratorFramework());
        }
        log.warn(ERR_MSG_SKIP_DIFF, nameSpace, "fail to connect to zk.");
        return null;
    }

    private List<TaskDifferentInfo> getTaskIdOnlyInZooKeeper(String nameSpace, Set<String> taskIdSetInDb) throws RuntimeException {
        List<TaskDifferentInfo> taskDiffInfoList = Lists.newArrayList();
        List<String> taskIdSetInZooKeeper = taskService.getTaskIdFromZooKeeper(nameSpace);
        for (String taskIdInZooKeeper : taskIdSetInZooKeeper) {
            if (taskIdSetInDb == null || taskIdSetInDb.isEmpty() || !taskIdSetInDb.contains(taskIdInZooKeeper)) {
                taskDiffInfoList.add(new TaskDifferentInfo(nameSpace, taskIdInZooKeeper, DifferentType.ZOOKEEPER_ONLY, Lists.<TaskConfigurationDifferentInfo>newArrayList()));
            }
        }
        return taskDiffInfoList;
    }

    private Set<String> getTaskIdSetInDb(List<TaskConfigurationDto> taskConfigurationDtoList) {
        Set<String> taskIdSet = Sets.newHashSet();
        for (TaskConfigurationDto taskConfigurationDto : taskConfigurationDtoList) {
            taskIdSet.add(taskConfigurationDto.getTaskId());
        }

        return taskIdSet;
    }

}
